#include "../main.h"

pthread_mutex_t file_mutex;
pthread_mutex_t msg_mutex;
pthread_mutex_t p_msg_mutex;
Queue fileQueue;
Queue msgQueue;
long timestart = 0;
long timeend = 0;

Map *msgmap;

void tidyMsg(RecvData *recvData)
{
    if (recvData->length > 0)
    {
        int a = 0;
        int result = 0;

        for (a = 0; a < recvData->length - 8; a++)
        {
            if (recvData->data[a] == 0x76)
            {
                if (recvData->data[a + 4] == 0x01 && recvData->data[a + 8] == 0x02)
                {
                    result = 1;
                    break;
                }
            }
        }
        if (result > 0)
        {
            if (a == 3)
            {
                return;
            }
            else
            {
                int leftsize = recvData->length - a + 3;
                char *lbuffer = (char *)malloc(leftsize);
                memset(lbuffer, 0, leftsize);
                memcpy(lbuffer + 3, recvData->data + a, recvData->length - a);
                free(recvData->data);
                recvData->data = NULL;
                recvData->data = lbuffer;
                recvData->length = leftsize;
                printf("aindex:%d\n,leftsize:%d,head:%02x\n", a, leftsize, recvData->data[3]);
            }
        }
    }
}

void put_into_queue(MNode *curNode)
{
    RecvData *recvData = curNode->record->recvData;
    if (curNode->record->recvData->length == 0)
    {
        return;
    }
    int get_head = can_get_head(recvData);
    while (get_head)
    {
        usleep(1000);
        tidyMsg(recvData);
        int msgsize = get_size(recvData->data + 3 * sizeof(int)) + 2 * sizeof(int);
        if (recvData->length >= msgsize + 4 * sizeof(int))
        {
            char *buffer = (char *)malloc(msgsize + 1);
            memset(buffer, 0, msgsize + 1);
            memcpy(buffer, recvData->data + 4 * sizeof(int), msgsize);
            buffer[msgsize] = '\0';
            parse_msg(curNode->record->sockfd, buffer, msgsize + 1);
            free(buffer);
            buffer = NULL;
            int bodysize = msgsize + 4 * sizeof(int);
            int leftsize = recvData->length - bodysize;
            if (leftsize > 0)
            {
                char *lbuffer = (char *)malloc(leftsize);
                memcpy(lbuffer, recvData->data + bodysize, leftsize);
                free(recvData->data);
                recvData->data = NULL;
                recvData->data = charCopy(lbuffer, leftsize);
                recvData->length = leftsize;
                free(lbuffer);
                lbuffer = NULL;
            }
            else
            {
                free(recvData->data);
                recvData->data = NULL;
                recvData->length = 0;
            }
        }
        else
        {
            // OUT_DEBUG(DEBUG, "#########can not parse node total:%d,real length:%d", msgsize + 16, recvData->length);
            int ret = tidy_width_otherHead(recvData);
            if (ret == 0)
            {
                break;
            }
            else
            {
                OUT_DEBUG(DEBUG, "###pre data can not find parse,but find other head #####");
            }
        }
        get_head = can_get_head(recvData);
        if (get_head == 0)
        {
            break;
        }
    }
}

void *process_data(void *args)
{
    RecvData *data = (RecvData *)args;
    addToMap(msgmap, data->sockfd, data->data, data->length);
    MNode *curNode = getbykey(msgmap, data->sockfd);
    RecvData *recvData = curNode->record->recvData;
    if (curNode->record->recvData->length > 0)
    {
        int get_head = can_get_head(recvData);
        if (get_head == 1)
        {
            put_into_queue(curNode);
        }
    }
    free(data->data);
    data->data = NULL;
    free(args);
    args = NULL;
    return NULL;
}

//添加原始消息
void appendData(int sockfd, char *msg, int len)
{
    if (len > 0)
    {
        if (msgmap == NULL)
        {
            msgmap = (Map *)malloc(sizeof(Map));
            msgmap->head = NULL;
        }
        RecvData *recvData = (RecvData *)malloc(sizeof(RecvData));
        recvData->length = len;
        recvData->data = charCopy(msg, len);
        recvData->sockfd = sockfd;
        pthread_t psend;
        if (pthread_create(&psend, NULL, process_data, (void *)recvData) != 0)
        {
            OUT_DEBUG(DEBUG, "create_thread error");
        }
        pthread_join(psend, NULL);
    }
}

//处理消息
void enterMsgQueue(char *data, int length, int id)
{
    parse_msg(id, data, length);
}

void *processMsgQueue()
{
    while (1)
    {
        pthread_mutex_lock(&p_msg_mutex);
        usleep(10);
        if (msgQueue.size > 0)
        {
            Msgbody *msg = NULL;
            msg = DeQueue(&msgQueue);
            if (msg != NULL)
            {
                if (msg->length > 0)
                {
                    parse_msg(msg->id, msg->data, msg->length);
                }
                free(msg->data);
                msg->data = NULL;
                free(msg);
                msg = NULL;
            }
        }
        pthread_mutex_unlock(&p_msg_mutex);
    }
}

int tidy_width_otherHead(RecvData *recvData)
{
    if (recvData->length > 0)
    {
        int a = 16;
        int result = 0;
        for (a = 16; a < recvData->length - 8; a++)
        {
            if (recvData->data[a] == 0x76)
            {
                if (recvData->data[a + 4] == 0x01 && recvData->data[a + 8] == 0x02)
                {
                    result = 1;
                    break;
                }
            }
        }
        if (result == 1)
        {
            int leftsize = recvData->length - a + 3;
            char *lbuffer = (char *)malloc(leftsize);
            memset(lbuffer, 0, leftsize);
            memcpy(lbuffer + 3, recvData->data + a, recvData->length - a);
            free(recvData->data);
            recvData->data = NULL;
            recvData->data = lbuffer;
            recvData->length = leftsize;
            printf("aindex:%d\n,leftsize:%d,head:%02x\n", a, leftsize, recvData->data[3]);
            return 1;
        }
    }
    return 0;
}

void *processData()
{
}

void doProcessMsg()
{
    InitQueue(&msgQueue);
    pthread_t t[3];
    pthread_mutex_init(&p_msg_mutex, NULL);
    int a = 0;
    for (a = 0; a < 3; a++)
    {
        if (pthread_create(&t[a], NULL, processMsgQueue, NULL) == -1)
        {
            exit(1);
        }
    }
}

void doMsg()
{
    pthread_t t[5];
    pthread_mutex_init(&msg_mutex, NULL);
    int a = 0;
    for (a = 0; a < 5; a++)
    {
        if (pthread_create(&t[a], NULL, processData, NULL) == -1)
        {
            exit(1);
        }
    }
}

int workThread()
{
    doMsg();
    doProcessMsg();
    //doFileMsg();
    return 0;
}